from pyflink.common import SimpleStringSchema, WatermarkStrategy, Time
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors.kafka import KafkaSource, KafkaOffsetsInitializer
from pyflink.datastream.window import ProcessingTimeSessionWindows

env = StreamExecutionEnvironment.get_execution_environment()

env.set_parallelism(1)

source = KafkaSource.builder().set_bootstrap_servers("master:9092").set_topics("bigdata").set_group_id(
    "my-group").set_starting_offsets(KafkaOffsetsInitializer.latest()).set_value_only_deserializer(
    SimpleStringSchema()).build()

lines_ds = env.from_source(source, WatermarkStrategy.no_watermarks(), "Kafka Source")

kv_ds = lines_ds.flat_map(lambda x: x.split(",")).map(lambda x: (x, 1))

key_by_ds = kv_ds.key_by(lambda x: x[0])

window_ds = key_by_ds.window(ProcessingTimeSessionWindows.with_gap(Time.seconds(10)))

count_ds = window_ds.reduce(lambda value1, value2: (value1[0], value1[1] + value2[1]))

count_ds.print()

env.execute()